use common::base::id_generator::gen_id_with_sgement_id;
use common::serial::api_descriptor::ServiceApiDescriptor;
use common::service::sys_service_api::ServiceKey;
use pyo3::pyfunction;
use pyo3::PyResult;
use std::collections::HashMap;
use std::mem::MaybeUninit;
use std::sync::Arc;
use structopt::StructOpt;

//
// 自身的描述内容
pub static mut DESCRIPTOR: MaybeUninit<ServiceApiDescriptor> = MaybeUninit::uninit();
// 消息分发
pub static mut IMPORTSERVICE_DESCRIPTOR: MaybeUninit<
    HashMap<String, Arc<Box<ServiceApiDescriptor>>>,
> = MaybeUninit::uninit();

pub static mut BOOT_ARGS: MaybeUninit<BootArgs> = MaybeUninit::uninit();

pub static mut SELF_SERVICE_KEY: MaybeUninit<ServiceKey> = MaybeUninit::uninit();

//
pub mod xport_future {
    use common::{
        base::status::Status,
        serial::request_message::RequestMessage,
        service::sys_service_api::{
            ChannelId, ConnIds, HTTPMethod, HttpRequest, HttpResponse, ServiceKey,
        },
    };
    use lazy_static::lazy_static;
    use pyo3::{
        exceptions::PyValueError,
        pyclass, pymethods,
        types::{IntoPyDict, PyDict},
        Py, PyResult, Python,
    };
    use std::{
        collections::HashMap,
        future::Future,
        sync::{Arc, Mutex},
        task::{Poll, Waker},
    };

    pub struct ResultApiState<T> {
        is_finish: bool,
        status: Status,
        value: Option<T>,
        waker: Option<Waker>,
    }

    pub struct ResultApiFuture<T> {
        pub shared_state: Arc<Mutex<ResultApiState<T>>>,
    }

    impl<T: Send + Sync> Future for ResultApiFuture<T> {
        type Output = Result<T, Status>;
        fn poll(
            self: std::pin::Pin<&mut Self>,
            cx: &mut std::task::Context<'_>,
        ) -> Poll<Self::Output> {
            let mut shared_state = self.shared_state.lock().unwrap();
            if shared_state.is_finish {
                if shared_state.status.is_erorr() {
                    let status = std::mem::take(&mut shared_state.status);
                    return Poll::Ready(Err(status));
                } else {
                    let value = std::mem::take(&mut shared_state.value);
                    return Poll::Ready(Ok(value.unwrap()));
                }
            }
            shared_state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }

    pub struct ImportApiState {
        pub is_finish: bool,
        pub status: Status,
        pub value: Option<Vec<u8>>,
        pub waker: Option<Waker>,
    }

    pub struct ImportApiFuture {
        pub shared_state: Arc<Mutex<ImportApiState>>,
    }

    impl Future for ImportApiFuture {
        type Output = Result<Vec<u8>, Status>;
        fn poll(
            self: std::pin::Pin<&mut Self>,
            cx: &mut std::task::Context<'_>,
        ) -> Poll<Self::Output> {
            let mut shared_state = self.shared_state.lock().unwrap();
            if shared_state.is_finish {
                if shared_state.status.is_erorr() {
                    let status = std::mem::take(&mut shared_state.status);
                    return Poll::Ready(Err(status));
                } else {
                    let value = std::mem::take(&mut shared_state.value);
                    return Poll::Ready(Ok(value.unwrap()));
                }
            }
            shared_state.waker = Some(cx.waker().clone());
            Poll::Pending
        }
    }

    //

    #[pyclass]
    pub struct PyConnIds {
        pub conn_id: Vec<i64>,
    }

    #[pyclass]
    pub struct PyChannelId {
        pub channel_id: Option<i64>,
    }

    #[pyclass]
    #[derive(Clone, Debug)]
    pub struct PyServiceKey {
        pub dxc_name: String,
        pub dxc_version: String,
        pub service_name: String,
    }

    #[derive(Clone, Debug)]
    #[pyclass]
    pub struct PyHttpRequest {
        pub method: Option<HTTPMethod>,
        pub url: String,
        pub headers: std::collections::HashMap<String, String>,
        pub body: String,
    }

    #[pymethods]
    impl PyServiceKey {
        #[new]
        pub fn new() -> PyServiceKey {
            PyServiceKey {
                dxc_name: String::default(),
                dxc_version: String::default(),
                service_name: String::default(),
            }
        }

        #[getter]
        pub fn dxc_name(&self) -> &String {
            &self.dxc_name
        }

        #[setter]
        pub fn set_dxc_name(&mut self, value: String) {
            self.dxc_name = value;
        }
        #[getter]
        pub fn dxc_version(&self) -> &String {
            &self.dxc_version
        }

        #[setter]
        pub fn set_dxc_version(&mut self, value: String) {
            self.dxc_version = value;
        }

        #[getter]
        pub fn service_name(&self) -> &String {
            &self.service_name
        }

        #[setter]
        pub fn set_service_name(&mut self, value: String) {
            self.service_name = value;
        }
    }

    impl PyServiceKey {
        pub fn into(service_key: ServiceKey) -> PyServiceKey {
            PyServiceKey {
                dxc_name: service_key.dxc_name,
                dxc_version: service_key.dxc_version,
                service_name: service_key.service_name,
            }
        }
        pub fn to(self) -> ServiceKey {
            let mut service_key = ServiceKey::default();
            service_key.dxc_name = self.dxc_name;
            service_key.dxc_version = self.dxc_version;
            service_key.service_name = self.service_name;
            service_key
        }
    }

    #[pymethods]
    impl PyHttpRequest {
        #[new]
        pub fn new() -> PyHttpRequest {
            PyHttpRequest {
                method: Some(HTTPMethod::GET),
                url: String::default(),
                headers: HashMap::new(),
                body: String::default(),
            }
        }
        #[getter]
        pub fn method(&self) -> Option<&'static str> {
            self.method.map(|m| match m {
                HTTPMethod::GET => "GET",
                HTTPMethod::POST => "POST",
            })
        }

        #[setter]
        pub fn set_method(&mut self, value: &str) -> PyResult<()> {
            match value {
                "GET" => self.method = Some(HTTPMethod::GET),
                "POST" => self.method = Some(HTTPMethod::POST),
                _ => return Err(PyValueError::new_err("Invalid HTTP method")),
            }
            Ok(())
        }

        #[getter]
        pub fn url(&self) -> &String {
            &self.url
        }

        #[setter]
        pub fn set_url(&mut self, value: String) {
            self.url = value;
        }

        #[getter]
        pub fn headers(&self) -> HashMap<String, String> {
            self.headers.clone()
        }

        #[setter]
        pub fn set_headers(&mut self, value: std::collections::HashMap<String, String>) {
            self.headers = value;
        }
        #[getter]
        pub fn body(&self) -> &String {
            &self.body
        }
        #[setter]
        pub fn set_body(&mut self, value: String) {
            self.body = value;
        }
    }

    #[pyclass]
    pub struct PyHttpResponse {
        pub headers: Py<PyDict>,
    }

    #[pymethods]
    impl PyHttpResponse {
        #[getter]
        pub fn headers(&self, py: Python) -> PyResult<Py<PyDict>> {
            Ok(self.headers.clone_ref(py))
        }
    }
    //
    impl PyHttpResponse {
        pub fn into(py: Python, resp: HttpResponse) -> PyHttpResponse {
            PyHttpResponse {
                headers: resp.headers.into_py_dict(py).into(),
            }
        }
    }

    impl PyConnIds {
        pub fn into(conn_ids: ConnIds) -> PyConnIds {
            PyConnIds {
                conn_id: conn_ids.conn_id,
            }
        }
    }

    impl PyChannelId {
        pub fn into(channel_id: ChannelId) -> PyChannelId {
            PyChannelId {
                channel_id: channel_id.channel_id,
            }
        }
    }

    impl PyHttpRequest {
        pub fn to(py_req: PyHttpRequest) -> HttpRequest {
            let mut http_req = HttpRequest::default();
            http_req.method = py_req.method;
            http_req.url = py_req.url;
            http_req.headers = py_req.headers;
            http_req.body = py_req.body;
            http_req
        }
    }
    //
    lazy_static! {
        static ref REQUEST_ID_MAP: Mutex<HashMap<i64, Box<dyn Fn(&[u8]) + Send + Sync>>> =
            Mutex::new(HashMap::new());
    }
    //
    pub fn add_request_handler(request_id: i64, handler: Box<dyn Fn(&[u8]) + Send + Sync>) {
        let mut handler_map = REQUEST_ID_MAP.lock().unwrap();
        handler_map.insert(request_id, handler);
    }

    pub fn build_result_api_future<T>() -> ResultApiFuture<T>
    where
        T: RequestMessage + Default + 'static,
    {
        let shared_state = Arc::new(Mutex::new(ResultApiState::<T> {
            is_finish: false,
            status: Status::default(),
            waker: None,
            value: None,
        }));
        let future = ResultApiFuture::<T> { shared_state };

        future
    }
    //

    pub fn add_request_result_handler<T>(
        request_id: i64,
        clone_shared_state: Arc<Mutex<ResultApiState<T>>>,
    ) where
        T: RequestMessage + Default + 'static,
    {
        add_request_handler(
            request_id,
            Box::new(move |buffer| {
                let waker = {
                    let mut shared_state = clone_shared_state.lock().unwrap();
                    //
                    let waker = shared_state.waker.take();
                    if waker.is_none() {
                        return;
                    }
                    let read_bytes = shared_state
                        .status
                        .parse_from_bytes_return_num(buffer)
                        .unwrap();

                    if read_bytes != buffer.len() as u64 {
                        let mut value = T::default();
                        let bytes = &buffer[read_bytes as usize..];
                        value.parse_from_bytes_return_num(bytes).unwrap();
                        shared_state.value = Some(value);
                    }
                    shared_state.is_finish = true;
                    waker
                };
                waker.unwrap().wake_by_ref();
            }),
        );
    }
    //

    pub fn result_in(request_id: i64, msg_body: &[u8]) {
        let handler = {
            let mut handler_map = REQUEST_ID_MAP.lock().unwrap();
            handler_map.remove(&request_id)
        };
        //
        if handler.is_none() {
            println!("返回消息: ={}，无处理....", request_id);
            return;
        }

        handler.unwrap()(msg_body);
        //
    }
    //
    pub fn build_import_api_future() -> ImportApiFuture {
        let shared_state = Arc::new(Mutex::new(ImportApiState {
            is_finish: false,
            status: Status::default(),
            waker: None,
            value: None,
        }));
        let future = ImportApiFuture { shared_state };
        future
    }
    //
}

#[derive(StructOpt, Debug)]
pub struct BootArgs {
    #[structopt(long)]
    pub dxc_name: String,

    #[structopt(long)]
    pub dxc_version: String,

    #[structopt(long)]
    pub service_name: String,

    #[structopt(long)]
    pub log_level: String,

    // 用户生成 id的
    #[structopt(long)]
    pub sgement_id: u16,
}

#[pyfunction]
pub fn service_name() -> PyResult<String> {
    unsafe { Ok(BOOT_ARGS.assume_init_ref().service_name.clone()) }
}

#[pyfunction]
pub fn gen_id() -> i64 {
    gen_id_inner()
}

fn gen_id_inner() -> i64 {
    unsafe {
        let boot_args = BOOT_ARGS.assume_init_ref();
        gen_id_with_sgement_id(boot_args.sgement_id)
    }
}

// 日志
pub mod log {
    use super::{ipc, BOOT_ARGS};
    use common::protocol::protocol_ipc::{self, ProtocolIPCWriter};
    use pyo3::{pyfunction, Python};
    use pyo3::{PyAny, PyResult};
    #[pyfunction]
    pub fn level() -> PyResult<String> {
        unsafe { Ok(BOOT_ARGS.assume_init_ref().log_level.clone()) }
    }

    #[pyfunction]
    pub fn output(py: Python, log_content: String) -> PyResult<&PyAny> {
        let log_content = log_content + "\n";
        pyo3_asyncio::tokio::future_into_py(py, async {
            output_to_xport(log_content).await;

            Ok(())
        })
    }

    async fn output_to_xport(log_content: String) {
        let request_id = super::gen_id_inner();
        let bytes: &[u8] = log_content.as_bytes();
        let mut ipc_writer = ProtocolIPCWriter::new(
            bytes.len() as u32,
            protocol_ipc::log_api::OUTPUT_API_ID,
            request_id,
        );
        ipc_writer.write_u8_msg_body(bytes);

        let msg_buffer = ipc_writer.msg_buffer;
        ipc::send_message(&msg_buffer).await;
    }
}

pub mod xport_api {
    use crate::serial::proto_to_pyobj::proto_to_pyobj;
    use crate::serial::pyobj_to_proto;
    use crate::x_api::xport_future::{add_request_handler, build_import_api_future};
    use crate::x_api::{ipc, service};

    use super::xport_future::{
        add_request_result_handler, build_result_api_future, PyChannelId, PyConnIds, PyServiceKey,
    };
    use super::{gen_id_inner, DESCRIPTOR, IMPORTSERVICE_DESCRIPTOR, SELF_SERVICE_KEY};
    use common::base::status::Status;
    use common::protocol::protocol_ipc::xport_api::SEND_MESSAGE_API_ID;
    use common::serial::read_string;
    use common::serial::request_message::RequestMessage;
    use common::service::sys_service_api::{ChannelId, ServiceKey};
    use common::{
        protocol::protocol_ipc::{self, ProtocolIPCWriter},
        service::sys_service_api::ConnIds,
    };
    use protobuf::CodedOutputStream;
    use pyo3::types::PyType;
    use pyo3::{pyfunction, Py, PyAny, PyErr, PyResult, Python};

    #[pyfunction]
    pub fn send_message<'a>(
        py: Python<'a>,
        dxc_name: &'a str,
        dxc_version: &'a str,
        service_name: &'a str,
        msg: &'a str,
        param: &'a PyAny,
    ) -> PyResult<&'a PyAny> {
        let dxc_key = format!("{}.{}", &dxc_name, &dxc_version);
        let type_type = py.get_type::<PyType>();

        let descriptor = unsafe {
            let descriptor = IMPORTSERVICE_DESCRIPTOR.assume_init_ref().get(&dxc_key);

            descriptor
        };
        if descriptor.is_none() {
            return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                "{} 描述信息不存在",
                dxc_key
            )));
        }
        let descriptor = descriptor.unwrap();
        let method_descriptor = descriptor.method(msg);
        if method_descriptor.is_none() {
            return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(format!(
                "{} 接口 {} 描述信息不存在",
                dxc_key, msg
            )));
        }
        let method_descriptor = method_descriptor.unwrap();

        // 获取消息头的i奶希
        let mut msg_size = protobuf::rt::string_size(1, msg);
        // let output_type = method_descriptor.output_type.as_ref().unwrap();
        let input_type = method_descriptor.input_type.as_ref().unwrap();

        let input_type_empty = input_type == ".google.protobuf.Empty";
        let mut receiver = ServiceKey::default();
        //
        receiver.dxc_name = dxc_name.to_owned();
        receiver.dxc_version = dxc_version.to_owned();
        receiver.service_name = service_name.to_owned();
        msg_size += receiver.compute_size_with_tag_and_len();

        if !input_type_empty {
            let result = pyobj_to_proto::compute_size(param, input_type, descriptor, type_type);
            if result.is_err() {
                return Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                    result.err().unwrap().err_msg,
                ));
            }
            // 写入标签
            msg_size += result.unwrap();
        }
        let request_id = gen_id_inner();

        // println!("receiver = {:?} request_id = {}", receiver, request_id);
        //
        let mut ipc_writer =
            ProtocolIPCWriter::new(msg_size as u32, SEND_MESSAGE_API_ID, request_id);
        let msg_body = ipc_writer.msg_body_buffer();

        let mut os = protobuf::CodedOutputStream::bytes(msg_body);
        //
        //
        receiver.serial_with_tag_and_len(&mut os);
        os.write_string(1, msg).unwrap();
        if !input_type_empty {
            pyobj_to_proto::pyobj_to_proto_with_buffer(
                param,
                &input_type,
                descriptor,
                type_type,
                &mut os,
            )
            .unwrap();
        }
        drop(os);
        let buffer = ipc_writer.msg_buffer;

        let msg = msg.to_owned();
        let dxc_name = dxc_name.to_owned();
        let dxc_version = dxc_version.to_owned();
        pyo3_asyncio::tokio::future_into_py(py, async move {
            let future = build_import_api_future();
            let clone_shared_state = future.shared_state.clone();
            add_request_handler(
                request_id,
                Box::new(move |buffer| {
                    let waker = {
                        let mut shared_state = clone_shared_state.lock().unwrap();
                        //
                        let waker = shared_state.waker.take();
                        if waker.is_none() {
                            return;
                        }

                        let mut is = protobuf::CodedInputStream::from_bytes(buffer);
                        let _tag = is.read_raw_tag_or_eof().unwrap();
                        let _rsp_msg = is.read_string();

                        println!("rsp_msg = {:?}", _rsp_msg);

                        shared_state
                            .status
                            .parse_from_input_stream_with_tag_and_len(&mut is);

                        if is.pos() != buffer.len() as u64 {
                            let bytes = &buffer[is.pos() as usize..];
                            shared_state.value = Some(bytes.to_vec());
                        }

                        shared_state.is_finish = true;
                        waker
                    };
                    waker.unwrap().wake_by_ref();
                }),
            );

            ipc::send_message(&buffer).await;

            let result = future.await;

            if let Ok(result) = result {
                Python::with_gil(move |py| {
                    let descriptor = unsafe {
                        let descriptor = IMPORTSERVICE_DESCRIPTOR.assume_init_ref().get(&dxc_key);

                        descriptor
                    };

                    let descriptor = descriptor.unwrap();
                    let method_descriptor = descriptor.method(&msg).unwrap();

                    let output_type = method_descriptor.output_type.as_ref().unwrap();
                    //
                    //
                    let py_module = service::get_import_module(&py, &dxc_name, &dxc_version);

                    let py_obj = proto_to_pyobj(&result, output_type, descriptor, py_module, &py);

                    if let Ok(py_obj) = py_obj {
                        Ok(py_obj)
                    } else {
                        Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                            py_obj.err().unwrap().err_msg,
                        ))
                    }
                })
            } else {
                Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                    result.err().unwrap().err_msg,
                ))
            }
        })
        // })
    }

    #[pyfunction]
    pub fn resopnse_message(
        py: Python,
        request_id: i64,
        _receiver: PyServiceKey,
        msg: &str,
        err_code: i64,
        err_msg: String,
        param: &PyAny,
    ) {
        let type_type = py.get_type::<PyType>();
        let (status, message) = if err_code == 0 {
            unsafe {
                let descriptor = DESCRIPTOR.assume_init_ref();
                let method_descriptor = descriptor.method(msg);
                if method_descriptor.is_none() {
                    return;
                }
                //
                let method_descriptor = method_descriptor.unwrap();
                let output_type = method_descriptor.output_type();
                //
                let result =
                    pyobj_to_proto::pyobj_to_proto(param, output_type, descriptor, type_type);
                if result.is_err() {
                    println!("py err = {:?}", result.err().unwrap());
                    return;
                }
                //
                let buffer = result.unwrap();
                (Status::default(), buffer)
            }
        } else {
            (Status::error(err_msg), Vec::new())
        };
        let rsp_message = format!("{}Rsp", msg);
        let mut size = protobuf::rt::string_size(1, &rsp_message);
        size += status.compute_size_with_tag_and_len();
        let message_slice = message.as_slice();
        let message_len = message_slice.len() as u64;
        if message_len != 0 {
            size += protobuf::rt::uint32_size(1, message_len as u32);
            size += message_len;
            // 消息体，要加上  tag 标签
        }
        //
        let mut ipc_writer = ProtocolIPCWriter::new(
            size as u32,
            protocol_ipc::xport_api::RESP_MESSAGE_API_ID,
            request_id,
        );
        let msg_body = ipc_writer.msg_body_buffer();

        let mut os = protobuf::CodedOutputStream::bytes(msg_body);
        os.write_string(1, &rsp_message).unwrap();
        status.serial_with_tag_and_len(&mut os);
        //
        if message_len != 0 {
            os.write_uint32(1, message_len as u32).unwrap();
            os.write_raw_bytes(message_slice).unwrap();
        }
        os.flush().unwrap();
        drop(os);
        //

        let msg_buffer = ipc_writer.msg_buffer;
        pyo3_asyncio::tokio::get_runtime().spawn(async move {
            ipc::send_message(&msg_buffer).await;
        });
    }
    //
    #[pyfunction]
    pub fn get_all_conn_id(py: Python) -> PyResult<&PyAny> {
        pyo3_asyncio::tokio::future_into_py(py, async {
            let request_id = gen_id_inner();
            let future = build_result_api_future::<ConnIds>();

            let clone_shared_state = future.shared_state.clone();
            add_request_result_handler(request_id, clone_shared_state);
            //

            let ipc_writer = ProtocolIPCWriter::new(
                0,
                protocol_ipc::xport_api::GET_ALL_CONN_ID_API_ID,
                request_id,
            );
            ipc::send_message(&ipc_writer.msg_buffer).await;
            let result = future.await;
            if let Ok(result) = result {
                Ok(PyConnIds::into(result))
            } else {
                Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                    result.err().unwrap().err_msg,
                ))
            }
        })
    }
    #[pyfunction]
    pub fn get_channel_id_by_conn_id(py: Python, conn_id: i64) -> PyResult<&PyAny> {
        pyo3_asyncio::tokio::future_into_py(py, async move {
            let request_id = gen_id_inner();
            let future = build_result_api_future::<ChannelId>();
            let clone_shared_state = future.shared_state.clone();
            add_request_result_handler(request_id, clone_shared_state);
            //

            let mut ipc_writer = ProtocolIPCWriter::new(
                8,
                protocol_ipc::xport_api::GET_CHANNEL_ID_BY_CONN_API_ID,
                request_id,
            );

            {
                let msg_buffer = ipc_writer.msg_body_buffer();
                let mut os = CodedOutputStream::bytes(msg_buffer);
                os.write_sfixed64_no_tag(conn_id).unwrap();
            }

            ipc::send_message(&ipc_writer.msg_buffer).await;

            let result = future.await;
            if let Ok(result) = result {
                Ok(PyChannelId::into(result))
            } else {
                Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                    result.err().unwrap().err_msg,
                ))
            }
        })
    }
}

pub mod http_api {
    use common::{
        protocol::protocol_ipc::{self, ProtocolIPCWriter},
        serial::request_message::RequestMessage,
        service::sys_service_api::HttpResponse,
    };
    use protobuf::CodedOutputStream;
    use pyo3::{pyfunction, PyAny, PyErr, PyResult, Python};

    use crate::x_api::ipc;
    use crate::xport_future::PyHttpRequest;

    use super::{
        gen_id_inner,
        xport_future::{add_request_result_handler, build_result_api_future, PyHttpResponse},
    };

    #[pyfunction]
    pub fn send_http_request(py: Python, py_http_req: PyHttpRequest) -> PyResult<&PyAny> {
        pyo3_asyncio::tokio::future_into_py(py, async {
            let request_id = gen_id_inner();
            let future = build_result_api_future::<HttpResponse>();
            let clone_shared_state = future.shared_state.clone();
            add_request_result_handler(request_id, clone_shared_state);
            //
            let http_req = PyHttpRequest::to(py_http_req);
            let size = http_req.compute_size();
            let mut ipc_writer = ProtocolIPCWriter::new(
                size as u32,
                protocol_ipc::http_api::SEND_HTTP_REQUEST_API_ID,
                request_id,
            );
            {
                let msg_buffer = ipc_writer.msg_body_buffer();
                let mut os = CodedOutputStream::bytes(msg_buffer);
                http_req.serial_with_output_stream(&mut os).unwrap();
            }

            ipc::send_message(&ipc_writer.msg_buffer).await;
            let result = future.await;
            if let Ok(result) = result {
                Python::with_gil(|py| Ok(PyHttpResponse::into(py, result)))
            } else {
                Err(PyErr::new::<pyo3::exceptions::PyValueError, _>(
                    result.err().unwrap().err_msg,
                ))
            }
        })
    }
}
pub mod service {
    use common::service::sys_service_api::ServiceKey;
    use protobuf::CodedInputStream;
    use pyo3::{
        types::{PyList, PyModule},
        Py, PyAny, PyErr, Python,
    };
    use std::mem::MaybeUninit;

    use crate::serial::proto_to_pyobj::proto_to_pyobj_by_is;

    use super::{xport_future::PyServiceKey, DESCRIPTOR};

    pub static mut DISPATCH_MESSAGE: MaybeUninit<Py<PyAny>> = MaybeUninit::uninit();

    pub static mut SOURCE_MODULE: MaybeUninit<Py<PyAny>> = MaybeUninit::uninit();

    pub fn first_connect_to_xport() {
        init();
        on_init();
    }

    fn get_source_module<'a>(py: &Python<'a>) -> &'a PyModule {
        let sys_module = py.import("sys").unwrap();
        let syspath: &PyList = match sys_module
            .getattr("path")
            .and_then(|p| p.downcast::<PyList>().map_err(|e| PyErr::from(e)))
        {
            Ok(path) => path,
            Err(_e) => {
                panic!("获取 sys.path 失败");
            }
        };
        syspath.insert(0, "./src/x_com").unwrap();
        py.import("source_api").unwrap()
    }

    pub fn get_import_module<'a>(
        py: &Python<'a>,
        dxc_name: &str,
        dxc_version: &str,
    ) -> &'a PyModule {
        let sys_module = py.import("sys").unwrap();
        let syspath: &PyList = match sys_module
            .getattr("path")
            .and_then(|p| p.downcast::<PyList>().map_err(|e| PyErr::from(e)))
        {
            Ok(path) => path,
            Err(_e) => {
                panic!("获取 sys.path 失败");
            }
        };
        syspath.insert(0, "./src/x_com/import_api").unwrap();

        let dxc_key = format!("{}_{}", dxc_name, dxc_version.replace(".", "_"));

        py.import(dxc_key.as_str()).unwrap()
    }

    pub fn init() {
        Python::with_gil(|py| {
            let sys_module = py.import("sys").unwrap();
            let syspath: &PyList = match sys_module
                .getattr("path")
                .and_then(|p| p.downcast::<PyList>().map_err(|e| PyErr::from(e)))
            {
                Ok(path) => path,
                Err(_e) => {
                    panic!("获取 sys.path 失败");
                }
            };
            syspath.insert(0, "./src/x_com").unwrap();
            let xport_core = py.import("xport_core").unwrap();
            let disptach_message_fn = xport_core.getattr("disptach_message").unwrap();
            let disptach_message_fn = Py::from(disptach_message_fn);
            unsafe {
                DISPATCH_MESSAGE.as_mut_ptr().write(disptach_message_fn);
            };
            //
            let source_module = get_source_module(&py);
            unsafe {
                SOURCE_MODULE.as_mut_ptr().write(Py::from(source_module));
            };
        });
    }

    pub fn on_init() {
        Python::with_gil(|py| unsafe {
            let py_service_key = PyServiceKey::new();

            DISPATCH_MESSAGE
                .assume_init_ref()
                .call1(py, (1, py_service_key, "on_init", py.None()))
                .unwrap();
        });
    }

    pub fn call_api(request_id: i64, msg_name: &str, param: Py<PyAny>) {
        Python::with_gil(|py| unsafe {
            let py_service_key = PyServiceKey::new();

            DISPATCH_MESSAGE
                .assume_init_ref()
                .call1(py, (request_id, py_service_key, msg_name, param))
                .unwrap();
        });
    }

    pub fn call_api_with_is(
        request_id: i64,
        sender_service_key: ServiceKey,
        msg_name: &str,
        input_stream: &mut CodedInputStream,
    ) {
        Python::with_gil(|py| unsafe {
            let descriptor = DESCRIPTOR.assume_init_ref();
            let method_descriptor = descriptor.method(&msg_name);
            if method_descriptor.is_none() {
                return;
            }

            let method_descriptor = method_descriptor.unwrap();
            let input_type = method_descriptor.input_type();
            let source_module = SOURCE_MODULE.assume_init_ref();
            let source_module = source_module.as_ref(py).downcast().unwrap();
            let result =
                proto_to_pyobj_by_is(&input_type, descriptor, source_module, input_stream, &py);
            if result.is_err() {
                println!("py err = {:?}", result.err().unwrap());
                return;
            }
            let py_sender_service_key = PyServiceKey::into(sender_service_key);
            DISPATCH_MESSAGE
                .assume_init_ref()
                .call1(
                    py,
                    (request_id, py_sender_service_key, msg_name, result.unwrap()),
                )
                .unwrap();
        });
    }
}

pub mod ipc {
    use lazy_static::lazy_static;
    use std::sync::Arc;

    use tokio::{
        io::{AsyncWriteExt, WriteHalf},
        net::windows::named_pipe::NamedPipeClient,
    };

    use crate::x_api::BOOT_ARGS;

    #[cfg(unix)]
    fn get_ipc_addr() -> String {
        unsafe {
            let boot_args = BOOT_ARGS.assume_init_ref();
            format!(
                "/xport/{}_{}_{}",
                boot_args.dxc_name, boot_args.dxc_version, boot_args.service_name
            )
        }
    }
    //
    lazy_static! {
        static ref IPC_WRITER: Arc<tokio::sync::Mutex<Option<WriteHalf<NamedPipeClient>>>> =
            Arc::new(tokio::sync::Mutex::new(None));
    }

    async fn set_write(_writer: WriteHalf<NamedPipeClient>) {
        let mut writer = IPC_WRITER.lock().await;

        *writer = Some(_writer);
    }

    pub async fn send_message(message: &Vec<u8>) -> bool {
        let mut writer = IPC_WRITER.lock().await;

        if let Some(writer) = writer.as_mut() {
            let ret = writer.write(message.as_slice()).await;

            if ret.is_err() {
                println!("发送消息失败, 长度：{}", message.len());
                return false;
            }
            let ret = writer.flush().await;

            ret.is_ok()
        } else {
            false
        }
    }

    /**
     * 连接到 xport
     */
    #[cfg(windows)]
    pub async fn connect_to_xport() {
        use crate::x_api::{service, xport_future};
        use common::protocol::protocol_v1::ProtocolV1Reader;
        use common::protocol::{get_version, MsgType};
        use protobuf::CodedInputStream;
        use std::ptr;
        use tokio::io::{self, AsyncReadExt};
        use tokio::net::windows::named_pipe::ClientOptions;
        let ipc_addr = unsafe {
            let boot_args = BOOT_ARGS.assume_init_ref();
            format!(
                r"\\.\pipe\\{}_{}_{}",
                boot_args.dxc_name, boot_args.dxc_version, boot_args.service_name
            )
        };
        //
        let mut try_connect_times = 0;
        //
        let mut is_first_connect = true;

        while try_connect_times < 3 {
            let client = ClientOptions::new()
                .open(&ipc_addr)
                .expect("连接 xport 失败！");
            println!("连接 ipc = {} 成功！", &ipc_addr);
            if is_first_connect {
                service::first_connect_to_xport();
                is_first_connect = false;
            }

            let (mut reader, writer) = io::split(client);
            // 设置 writer
            set_write(writer).await;

            let mut header = [0u8; 4];
            loop {
                let result = reader.read_exact(&mut header).await;
                if result.is_err() {
                    try_connect_times += 1;
                    break;
                }
                //
                let msg_size = u32::from_le_bytes(header);
                //
                let message_len = msg_size as usize;
                //
                let mut message: Vec<u8> = Vec::with_capacity(message_len);
                //
                unsafe {
                    message.set_len(message_len);
                    ptr::copy_nonoverlapping(header.as_ptr(), message.as_mut_ptr(), 4);
                }
                let msg_buffer = &mut message[4..];
                let result = reader.read_exact(msg_buffer).await;
                if result.is_err() {
                    break;
                }
                let slice = message.as_slice();

                let version = get_version(slice);
                if version != 1 {
                    println!("只支持 版本 1 的协议解析！");
                    return;
                }
                let reader = ProtocolV1Reader::new(slice);
                let msg_type = reader.msg_type();
                if msg_type.is_none() {
                    println!("读取 msg_type 失败！");
                    return;
                }
                let request_id = reader.req_id();
                let msg_body = reader.msg_body();
                //
                let msg_type = msg_type.unwrap();
                if msg_type == MsgType::RspNormalMsg {
                    //
                    xport_future::result_in(request_id, msg_body);
                } else {
                    //
                    let mut input_stream = CodedInputStream::from_bytes(msg_body);
                    //
                    let _tag = input_stream.read_raw_tag_or_eof().unwrap();
                    //
                    let message = input_stream.read_string().unwrap();

                    service::call_api_with_is(
                        request_id,
                        reader.receiver(),
                        &message,
                        &mut input_stream,
                    );
                }
            }
        }
    }
}
